//package com.huc.tools.kafka.consumer;
//
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.ArrayList;
//import java.util.Arrays;
//import java.util.List;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//
///**
// * @Author: 胡成
// * @Version: 0.0.1V
// * @Date: 2018/8/13
// * @Description: 类描述
// **/
//@Component
//public class ConsumerStart {
//
//    @Autowired
//    private KafkaConsumer<String,String> consumer;
//
//    public void start(){
//        consumer.subscribe(Arrays.asList("topic-test"));
//        while (true){
//            ConsumerRecords<String,String> records = consumer.poll(1000);
//            for (ConsumerRecord<String,String> record:records){
//                System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(),record.key(),record.value());
//            }
//        }
//    }
//
//    public void start2(){
//
//        ExecutorService service = Executors.newFixedThreadPool(3);
//        service.execute(()->{
//            System.out.println("kafka消费开始......");
//            // 设置消费组订阅的 Topic，可以订阅多个
//            // 如果 GROUP_ID_CONFIG 是一样，则订阅的 Topic 也建议设置成一样
//            List<String> subscribedTopics = new ArrayList<>();
//            // 如果需要订阅多个 Topic，则在这里 add 进去即可
//            // 每个 Topic 需要先在控制台进行创建
//            subscribedTopics.add("topic-test");
//            consumer.subscribe(subscribedTopics);
//            // 循环消费消息
//            while (true){
//                try {
//                    ConsumerRecords<String, String> records = consumer.poll(1000);
//                    // 必须在下次 poll 之前消费完这些数据, 且总耗时不得超过 SESSION_TIMEOUT_MS_CONFIG
//                    // 建议开一个单独的线程池来消费消息，然后异步返回结果
//                    for (ConsumerRecord<String, String> record : records) {
//                        System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(),record.key(),record.value());
//                    }
//                } catch (Exception e) {
//                    try {
//                        Thread.sleep(1000);
//                    } catch (Throwable ignore) {
//                    }
//                    e.printStackTrace();
//                }
//            }
//        });
//    }
//}
